package com.sqoop.server.receiver;

import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.sqoop.commons.receive.MsgListener;
import com.sqoop.commons.receive.impl.MqMsgListener;
import com.sqoop.core.destnation.MqMsgDestination;
import com.sqoop.core.msg.Msg;
import com.sqoop.core.request.Request;
import com.sqoop.server.BaseProcess;
import org.apache.commons.io.IOUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import sun.rmi.runtime.Log;

import java.io.File;
import java.io.IOException;
import java.util.List;

/**
 * Created by zw on 2017/11/10.
 */
public class MsgReceiverProcess extends BaseProcess{
    private static Logger logger = LogManager.getLogger(MsgReceiverProcess.class.getName());

    private MsgListener msgListener = null;
    private String configFile = "conf/mqconfig.json";
    private Thread msgReceiveThread = null;
    @Override
    public void process(Msg msg) {
        if (msg instanceof Request){
//            logger.info(new Gson().toJson(msg).toString());
//            System.out.println("parse a msg " +new Gson().toJson(msg).toString());
            aSend(msg);
        }else if (msg instanceof ReceiverControlMsg){
            ReceiverControlMsg controlMsg = (ReceiverControlMsg)msg;
        }
    }

    @Override
    public void init() {
        msgListener = new MqMsgListener();
        File file = new File(configFile);
        if (!file.exists()){
            return;
        }
        Resource resource = new FileSystemResource(configFile);
        try {
            String configStr = IOUtils.toString(resource.getInputStream());
            List<MqMsgDestination> mqMsgDestinations = new Gson().fromJson(configStr, new TypeToken<List<MqMsgDestination>>() {}.getType());
            for (MqMsgDestination mqMsgDestination:mqMsgDestinations){
                msgListener.register(mqMsgDestination);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        msgReceiveThread = new Thread(new MsgReceiveRunnable());
        msgReceiveThread.setName("receivethread");
        msgReceiveThread.start();
    }

    @Override
    public void stop0() {
        msgListener.stop();
    }

    @Override
    public void pause0() {
        msgListener.pause();
    }

    public class MsgReceiveRunnable implements Runnable{
        @Override
        public void run() {
            while (true) {
                Msg msg = msgListener.getMsg();
                addMsg(msg);
            }
        }
    }

    public static void main(String[] args) throws IOException {
        //String path = MsgReceiverProcess.class.getProtectionDomain().getCodeSource().getLocation().getPath() + "../" + "config/receiver.json";
        String path = "./conf/mqconfig.json";
        System.out.println(path);

        File file = new File(path);
        Resource resource = new FileSystemResource(file);
        String configStr = IOUtils.toString(resource.getInputStream());
        System.out.println(configStr);

        MsgReceiverProcess msgReceiverProcess = new MsgReceiverProcess();
        msgReceiverProcess.start();;

    }

}